package cn.yto.rocketmq.rocketmqselect.config;



import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.stereotype.Component;

import java.util.HashSet;
import java.util.List;
import java.util.Set;


/**
  * 消费前置库的mq 而后延迟发送到结算mq
  * @Author: myzf
  * @Date: 2020/8/17 14:38
  * @param
*/


@Component
public class FrontBuildCenterRocketMQDelayProduct {

	/** 消费前置库的mq地址*/
	@Value("${rmq.topicReceiveAddress}")
	private String topicReceiveAddress;
	/** 消费前置库的mq消费组*/
	@Value("${rmq.topicReceiveConsumerGroup}")
	private String topicReceiveConsumerGroup;
	/** 消费前置库的mq topic*/
	@Value("${rmq.receiveConsumerTopic}")
	private Set<String> receiveConsumerTopic = new HashSet<String>();
	// 消费最小线程数， 默认10

	private String threadMin = "20";

	private String threadMax = "30";

	private String consumeMessageBatchMaxSize = "100";

	private String pullBatchSize = "100";

	private String pullInterval = "0";

	private String persistConsumerOffsetInterval = "5000";

	private static final String TOPIC_TYPE = "*";


	/**
	 * 初始化RocketMQ消费者客户端信息
	 * 
	 * @return RocketMQ消费者客户端
	 * @throws MQClientException
	 */
	private DefaultMQPushConsumer initMQpushConsumer() throws MQClientException {
		// 同属一个应用设为同一个组名
		DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(topicReceiveConsumerGroup);
		for (String topic : receiveConsumerTopic) {
			// 订阅所有topic
			consumer.subscribe(topic, TOPIC_TYPE);
		}
		// 设置ip集群
		consumer.setNamesrvAddr(topicReceiveAddress);

		consumer.setInstanceName("ccc");

		// 设置消费最小线程数
		consumer.setConsumeThreadMin(Integer.valueOf(threadMin));
		// 设置消费最大线程数
		consumer.setConsumeThreadMax(Integer.valueOf(threadMax));
		// 设置消费集群模式
		consumer.setMessageModel(MessageModel.CLUSTERING);
		// 一次消费消息数
		consumer.setConsumeMessageBatchMaxSize(Integer.valueOf(consumeMessageBatchMaxSize));
		// 每次拉取消息数据
		consumer.setPullBatchSize(Integer.valueOf(pullBatchSize));
		// 拉消息间隔
		consumer.setPullInterval(Integer.valueOf(pullInterval));
		// 持久化Consumer消费进度到mq上间隔时间
		consumer.setPersistConsumerOffsetInterval(Integer.valueOf(persistConsumerOffsetInterval));
		//第一次启动对尾开始启动  默认
//		consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
		return consumer;
	}

	//@Bean
	public DefaultMQPushConsumer pushFrontConsumer() throws MQClientException {
		// 初始化消费组设置
		DefaultMQPushConsumer consumer = initMQpushConsumer();
		consumer.registerMessageListener(new MessageListenerConcurrently() {
			public ConsumeConcurrentlyStatus consumeMessage(final List<MessageExt> msgs,
                                                            final ConsumeConcurrentlyContext context) {
				return handlerMessage(msgs);
			}
		});
		consumer.start();
		return consumer;
	}

	/**
	 * 处理消费者事件
	 * 
	 * @param messageList
	 *            接收到的消息集合
	 * @return 处理是否成功，返回给生产者
	 */
	private ConsumeConcurrentlyStatus handlerMessage(List<MessageExt> messageList) {
		for (MessageExt msg : messageList) {
			try {

				byte[] body = msg.getBody();
				String s = new String(body);
				System.out.println(s);


			} catch (Exception e) {

			}
		}

		return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
	}




}
